package demo;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.ArrayList;
import java.util.HashMap;

public class MyConsumer2 {
    public static void main(String[] args) {
        // 需求：通过java,构建一个消费者，用于消费kafka中某个topic的数据
        // 1、构建一个map对象，存放具体的配置信息
        HashMap<String, Object> confMap = new HashMap<>();
        confMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop11:9092,hadoop12:9092,hadoop13:9092");
        confMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        confMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        // 没有组名，不让消费数据
        confMap.put(ConsumerConfig.GROUP_ID_CONFIG,"g1");

        // 2、构建一个consumer客户端用于接收消费者到的数据
        KafkaConsumer<String,String> consumerClient = new KafkaConsumer<>(confMap);
        // 不要急于拉取数据
        // 因为我们现在还不知道要拉取哪个topic的数据
        ArrayList<String> topics = new ArrayList<>();
        topics.add("topicB");
        consumerClient.subscribe(topics); // 同时消费多个topic

        while (true) {
            // 3、拉取数据
            // 一旦开始拉取数据，那么停不下来，如果数据已经消费完了，就开始进行等待，间隔时间查看topic是否有新数据
            ConsumerRecords<String, String> consumerRecords = consumerClient.poll(2000);
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord.topic() + "中的值为：" + consumerRecord.value() + "分区为" + consumerRecord.partition());
            }
        }

        // 4、关闭资源 （死循环后面不会有代码）
//        consumerClient.close();
    }
}
